/* Copyright (c) 2023 Renmin University of China
RMDB is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:
        http://license.coscl.org.cn/MulanPSL2
THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details. */

#pragma once
#include "execution_defs.h"
#include "execution_manager.h"
#include "executor_abstract.h"
#include "index/ix.h"
#include "system/sm.h"
#include <cstring>
#include <memory>
#include <vector>

class UpdateExecutor : public AbstractExecutor {
   private:
    TabMeta tab_;
    std::vector<Condition> conds_;
    RmFileHandle *fh_;
    std::vector<Rid> rids_;
    std::string tab_name_;
    std::vector<SetClause> set_clauses_;
    SmManager *sm_manager_;

   public:
    UpdateExecutor(SmManager *sm_manager, const std::string &tab_name, std::vector<SetClause> set_clauses,
                   std::vector<Condition> conds, std::vector<Rid> rids, Context *context) {
        sm_manager_ = sm_manager;
        tab_name_ = tab_name;
        set_clauses_ = set_clauses;
        tab_ = sm_manager_->db_.get_table(tab_name);
        fh_ = sm_manager_->fhs_.at(tab_name).get();
        conds_ = conds;
        rids_ = rids;
        context_ = context;
    }
    std::unique_ptr<RmRecord> Next() override {
        // 先获取对应的表锁
        // context_->lock_mgr_->lock_IX_on_table(context_->txn_, fh_->GetFd());
        context_->lock_mgr_->lock_exclusive_on_table(context_->txn_, fh_->GetFd());

        int record_size = fh_->get_file_hdr().record_size;
        auto buf = std::make_unique<char[]>(record_size);

        for (const auto &rid: rids_) {
            std::unique_ptr<RmRecord> record = fh_->get_record(rid, context_);
            memcpy(buf.get(), record->data, record_size);

            applySetClauses(buf.get(), record);

            RmRecord old_record{record->size};
            RmRecord new_record{record->size};

            memcpy(old_record.data, record->data, record->size);
            memcpy(new_record.data, buf.get(), record->size);

            checkForPrimaryKeyUpdate(old_record, new_record);

            fh_->update_record(rid, buf.get(), context_);
            updateIndexes(old_record, new_record, rid);

            WriteRecord *write_record = new WriteRecord(WType::UPDATE_TUPLE, tab_name_, rid, old_record);
            context_->txn_->append_write_record(write_record);
        }
        return nullptr;
    }

    Rid &rid() override { return _abstract_rid; }

private:
    void applySetClauses(char *buf, const std::unique_ptr<RmRecord>& record) {
        for (auto &clause: set_clauses_) {
            if(clause.is_val){
                auto col = tab_.get_col(clause.lhs.col_name);
                clause.rhs.init_raw(col->len);
                memcpy(buf + col->offset, clause.rhs.raw->data, col->len);
            }else{
                Value temp_value;
                auto lhs_col = tab_.get_col(clause.lhs.col_name);
                char *data = record->data + lhs_col->offset;
                switch (lhs_col->type)
                {
                    case TYPE_INT:
                    {
                        int sum = 0;
                        sum += *reinterpret_cast<int *>(data);
                        sum += clause.rhs.int_val;
                        // temp_value.set_int(sum);
                        memcpy(buf + lhs_col->offset, &sum, lhs_col->len);
                        break;
                    }
                    case TYPE_FLOAT:
                    {
                        float sum = 0;
                        sum += *reinterpret_cast<float *>(data);
                        sum += clause.rhs.float_val;
                        // temp_value.set_float(sum);
                        memcpy(buf + lhs_col->offset, &sum, lhs_col->len);
                        break;
                    }
                    default:
                    {
                        throw InternalError("Invalid Type");
                        break;
                    }
                }
                // temp_value.init_raw(lhs_col->len);
                // memcpy(buf + lhs_col->offset , temp_value.raw->data, lhs_col->len);
            }
        }
    }

    void checkForPrimaryKeyUpdate(const RmRecord &old_record, const RmRecord &new_record) {
        for (const auto &index : tab_.indexes) {
            auto ih = sm_manager_->ihs_.at(index.index_name).get();
            auto old_key = index.get_key(old_record.data);
            auto new_key = index.get_key(new_record.data);

            if (memcmp(old_key.get(), new_key.get(), index.col_tot_len) != 0) {
                std::vector<Rid> result;
                if (ih->get_value(new_key.get(), &result, context_->txn_)) {
                    throw DuplicateKeyError(index.index_name);
                }
            }
        }
    }

    void updateIndexes(const RmRecord &old_record, const RmRecord &new_record, const Rid &rid) {
        for (const auto &index : tab_.indexes) {
            if (isIndexUpdated(index)) {
                auto ih = sm_manager_->ihs_.at(index.index_name).get();
                auto delete_key = index.get_key(old_record.data);
                auto insert_key = index.get_key(new_record.data);

                ih->delete_entry(delete_key.get(), context_->txn_);
                ih->insert_entry(insert_key.get(), rid, context_->txn_);
            }
        }
    }

    bool isIndexUpdated(const IndexMeta &index) {
        const auto &cols = index.cols;
        for (const auto &set_clause : set_clauses_) {
            auto pos = std::find_if(cols.begin(), cols.end(), [&set_clause](const ColMeta &col) {
                return col.name == set_clause.lhs.col_name;
            });
            if (pos != cols.end()) {
                return true;
            }
        }
        return false;
    }
};